package com.zhanghe.study.zookeeper;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @author zh
 * @date 2023/7/28 14:45
 */
public class TestZkclient {

    static class StringZkSerializer implements ZkSerializer {

        @Override
        public byte[] serialize(Object data) throws ZkMarshallingError {
            return String.valueOf(data).getBytes(StandardCharsets.UTF_8);
        }

        @Override
        public Object deserialize(byte[] bytes) throws ZkMarshallingError {
            return new String(bytes,StandardCharsets.UTF_8);
        }
    }

    // 连接地址
    private final String connect = "localhost:2181";

    // 单位是毫秒
    private final int sessionTimeOut = 2000;
    private final int connectionTimeout = 2000;

    private ZkClient zkClient;

    @Before
    public void init() {
        // zkServers  连接地址，如果是集群的话则使用逗号拼接  如 "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
        // sessionTimeOut session过期时间，单位毫秒
        // connectionTimeout  连接创建超时时间，单位毫秒
        zkClient = new ZkClient(connect, sessionTimeOut,connectionTimeout,new StringZkSerializer());
    }

    // 创建节点
    @Test
    public void createNode() {

        // path    路径
        // data    数据

        // 创建模式，持久节点
        zkClient.createPersistent("/zkclient_test","第一次创建节点");


        // createParents为true 递归创建父节点
        zkClient.createPersistent("/p/c",true);


    }

    // 设置节点数据
    @Test
    public void setNodeData() {
        // path  路径
        // data  the data to set  数据
        // version  所期待的版本，相当于乐观锁，如果是-1，则匹配所有版本
        zkClient.writeData("/zkclient_test","修改节点数据",-1);

    }

    // 获取节点数据
    @Test
    public void getNodeData() {

        Stat stat = new Stat();
        //path the given path 路径
        //stat the stat of the node  获取状态信息
        String data = zkClient.readData("/zkclient_test",stat);
        System.out.println(data);
        System.out.println(stat);
    }

    // 获取子节点
    @Test
    public void getChildNodes() throws InterruptedException {
        // 第二个参数为监听器
        // zkClient的监听器不是一次性的，会一直有效
        zkClient.subscribeChildChanges("/test_watch", new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                System.out.println(parentPath+"中子节点发生变化，当前子节点列表为"+currentChilds);
            }
        });

        // 监听打印内容 /test_watch中子节点发生变化，当前子节点列表为[]
        zkClient.createPersistent("/test_watch");
        Thread.sleep(1000);
        System.out.println(zkClient.getChildren("/test_watch"));
        // 监听打印内容 /test_watch中子节点发生变化，当前子节点列表为[c1]
        zkClient.createPersistent("/test_watch/c1");
        Thread.sleep(1000);
        // 监听打印内容 /test_watch中子节点发生变化，当前子节点列表为[]
        zkClient.delete("/test_watch/c1");
        Thread.sleep(1000);
        // 监听打印内容 /test_watch中子节点发生变化，当前子节点列表为null
        zkClient.delete("/test_watch");
        Thread.sleep(3000);


    }

    // 删除节点
    @Test
    public void deleteNode() {
        // 使用deleteRecursive可以逐层删除节点
        zkClient.deleteRecursive("/p/c");
    }
}
